package cn.iotab.iotkit.transport.mqtt.handler;

import cn.iotab.iotkit.transport.api.driver.DriverSession;
import cn.iotab.iotkit.transport.mqtt.MqttTransportContext;
import cn.iotab.iotkit.transport.mqtt.session.MqttSessionCtx;
import cn.iotab.iotkit.transport.mqtt.session.MqttTopicMatcher;
import cn.iotab.iotkit.transport.mqtt.topic.MqttTopics;
import cn.iotab.iotkit.transport.mqtt.topic.TopicMeta;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class UnsubscribeHandler extends AbstractMqttMessageHandler<MqttUnsubscribeMessage>{
    public UnsubscribeHandler(MqttTransportContext context, MqttSessionCtx deviceSessionCtx) {
        super(context, deviceSessionCtx);
    }

    @Override
    public void process(ChannelHandlerContext ctx, MqttUnsubscribeMessage mqttMsg) {
        if (!checkConnected(ctx, mqttMsg)) {
            return;
        }
        log.trace("[{}] Processing unSubscription [{}]!", sessionId, mqttMsg.variableHeader().messageId());
        boolean unsubscribeMsgTopic = false;
        boolean unsubscribeReplyTopic = false;
        for (String topicName : mqttMsg.payload().topics()) {
            deviceSessionCtx.getMqttQoSMap().remove(new MqttTopicMatcher(topicName));
            //
            TopicMeta topicMeta = MqttTopics.parseTopic(topicName);
            switch (topicMeta.getTopicType()) {
                case PLATFORM_POST_DEVICE:
                    log.info("[{}] Unsubscribe topic {}", sessionId, topicName);
                    unregisterSubQoS(topicName);
                    unsubscribeMsgTopic = true;
                    break;
                case PLATFORM_REPLY_DEVICE:
                    log.info("[{}] Unsubscribe topic {}", sessionId, topicName);
                    unregisterSubQoS(topicName);
                    unsubscribeReplyTopic = true;
                    break;
                default:
                    log.warn("[{}] Failed to unsubscribe to [{}]: invalid topic", sessionId, topicName);
                    break;
            }
        }
        transportDriverService.sessionActivity(deviceSessionCtx.getSessionInfo());
        if (unsubscribeMsgTopic || unsubscribeReplyTopic){
            DriverSession sessionMD = deviceSessionCtx.getDriverSession();
            if (sessionMD != null){
                if (unsubscribeMsgTopic){
                    sessionMD.setSubscribedMessageTopic(false);
                }
                if (unsubscribeReplyTopic){
                    sessionMD.setSubscribedReplyTopic(false);
                }
            }
        }

        ctx.writeAndFlush(createUnSubAckMessage(mqttMsg.variableHeader().messageId()));
    }

    private MqttMessage createUnSubAckMessage(int msgId) {
        MqttFixedHeader mqttFixedHeader =
                new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttMessageIdVariableHeader mqttMessageIdVariableHeader = MqttMessageIdVariableHeader.from(msgId);
        return new MqttMessage(mqttFixedHeader, mqttMessageIdVariableHeader);
    }

    private void unregisterSubQoS(String topic) {
        deviceSessionCtx.getMqttQoSMap().remove(new MqttTopicMatcher(topic));
    }
}
